package ontime

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
  * @author dinghao 
  * @create 2021-07-28 11:50 
  * @message
  */
object Eventflink {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val socketdata = env.socketTextStream(args(0),9999)
    val mapdata = socketdata.map(a => {
      val data = a.split(" ")
      (data(0), data(1).toLong, data(2).toInt)
    })
    mapdata.keyBy(0)
      .process(
        new KeyedProcessFunction[Tuple,(String, Long, Int), String]{
          override def onTimer(timestamp: Long,
                               ctx: KeyedProcessFunction[Tuple, (String, Long, Int), String]#OnTimerContext,
                               out: Collector[String]): Unit = super.onTimer(timestamp, ctx, out)

          override def processElement(value: (String, Long, Int),
                                      ctx: KeyedProcessFunction[Tuple, (String, Long, Int), String]#Context,
                                      out: Collector[String]): Unit = {
            if(value._3>5) {
              println(ctx.getCurrentKey)
            }
            out.collect(value.toString())
          }
        }
      ).print("keyprocess:")
    env.execute("eventflink")
  }
}
